//
// Created by Administrator on 2025/9/25.
//

#ifndef ERRORDEMO_THREADSAFEQUEUE_H
#define ERRORDEMO_THREADSAFEQUEUE_H
#include <deque>
#include <mutex>
#include <condition_variable>
#include <optional>
#include <stdexcept>
#include <chrono>

template <typename T>
class ThreadSafeQueue {
public:
    ThreadSafeQueue(size_t init_capacity = 16,
                    bool expandable = true,
                    double expand_factor = 2.0,
                    size_t expand_step = 0)
        : capacity_(init_capacity),
          expandable_(expandable),
          expand_factor_(expand_factor),
          expand_step_(expand_step)
    {
        if (capacity_ == 0) throw std::invalid_argument("capacity must be > 0");
        if (expand_factor_ <= 1.0 && expand_step_ == 0 && expandable_) {
            throw std::invalid_argument("invalid expansion strategy");
        }
    }

    // 禁止拷贝
    ThreadSafeQueue(const ThreadSafeQueue&) = delete;
    ThreadSafeQueue& operator=(const ThreadSafeQueue&) = delete;

    // ------------------- push（阻塞直到能插入） -------------------
    void push(const T& value) {
        std::unique_lock<std::mutex> lock(mtx_);
        cv_not_full_.wait(lock, [this] { return queue_.size() < capacity_ || expandable_; });
        if (queue_.size() >= capacity_ && expandable_) {
            expand_capacity();
        }
        queue_.push_back(value);
        cv_not_empty_.notify_one();
    }

    void push(T&& value) {
        std::unique_lock<std::mutex> lock(mtx_);
        cv_not_full_.wait(lock, [this] { return queue_.size() < capacity_ || expandable_; });
        if (queue_.size() >= capacity_ && expandable_) {
            expand_capacity();
        }
        queue_.push_back(std::move(value));
        cv_not_empty_.notify_one();
    }

    // ------------------- push_for（带超时） -------------------
    template <class Rep, class Period>
    bool push_for(const T& value, const std::chrono::duration<Rep, Period>& timeout) {
        std::unique_lock<std::mutex> lock(mtx_);
        if (!cv_not_full_.wait_for(lock, timeout, [this] { return queue_.size() < capacity_ || expandable_; })) {
            return false; // 超时
        }
        if (queue_.size() >= capacity_ && expandable_) {
            expand_capacity();
        }
        queue_.push_back(value);
        cv_not_empty_.notify_one();
        return true;
    }

    template <class Rep, class Period>
    bool push_for(T&& value, const std::chrono::duration<Rep, Period>& timeout) {
        std::unique_lock<std::mutex> lock(mtx_);
        if (!cv_not_full_.wait_for(lock, timeout, [this] { return queue_.size() < capacity_ || expandable_; })) {
            return false; // 超时
        }
        if (queue_.size() >= capacity_ && expandable_) {
            expand_capacity();
        }
        queue_.push_back(std::move(value));
        cv_not_empty_.notify_one();
        return true;
    }

    // ------------------- try_push（非阻塞） -------------------
    bool try_push(const T& value) {
        std::lock_guard<std::mutex> lock(mtx_);
        if (queue_.size() >= capacity_ && !expandable_) {
            return false; // 队列已满且不可扩容
        }
        if (queue_.size() >= capacity_ && expandable_) {
            expand_capacity();
        }
        queue_.push_back(value);
        cv_not_empty_.notify_one();
        return true;
    }

    bool try_push(T&& value) {
        std::lock_guard<std::mutex> lock(mtx_);
        if (queue_.size() >= capacity_ && !expandable_) {
            return false; // 队列已满且不可扩容
        }
        if (queue_.size() >= capacity_ && expandable_) {
            expand_capacity();
        }
        queue_.push_back(std::move(value));
        cv_not_empty_.notify_one();
        return true;
    }

    // ------------------- pop（阻塞式获取） -------------------
    T pop() {
        std::unique_lock<std::mutex> lock(mtx_);
        cv_not_empty_.wait(lock, [this] { return !queue_.empty(); });
        T value = std::move(queue_.front());
        queue_.pop_front();
        cv_not_full_.notify_one();
        return value;
    }

    // ------------------- pop_for（带超时） -------------------
    template <class Rep, class Period>
    std::optional<T> pop_for(const std::chrono::duration<Rep, Period>& timeout) {
        std::unique_lock<std::mutex> lock(mtx_);
        if (!cv_not_empty_.wait_for(lock, timeout, [this] { return !queue_.empty(); })) {
            return std::nullopt; // 超时
        }
        T value = std::move(queue_.front());
        queue_.pop_front();
        cv_not_full_.notify_one();
        return value;
    }

    // ------------------- try_pop（非阻塞） -------------------
    std::optional<T> try_pop() {
        std::lock_guard<std::mutex> lock(mtx_);
        if (queue_.empty()) return std::nullopt;
        T value = std::move(queue_.front());
        queue_.pop_front();
        cv_not_full_.notify_one();
        return value;
    }

    bool empty() const {
        std::lock_guard<std::mutex> lock(mtx_);
        return queue_.empty();
    }

    size_t size() const {
        std::lock_guard<std::mutex> lock(mtx_);
        return queue_.size();
    }

    size_t capacity() const {
        std::lock_guard<std::mutex> lock(mtx_);
        return capacity_;
    }

private:
    void expand_capacity() {
        if (!expandable_) return;
        if (expand_step_ > 0) {
            capacity_ += expand_step_;
        } else {
            capacity_ = static_cast<size_t>(capacity_ * expand_factor_);
        }
    }

    mutable std::mutex mtx_;
    std::condition_variable cv_not_empty_;
    std::condition_variable cv_not_full_;
    std::deque<T> queue_;
    size_t capacity_;
    bool expandable_;
    double expand_factor_;
    size_t expand_step_;
};

#endif //ERRORDEMO_THREADSAFEQUEUE_H
